package com.xiaohu.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

//读取hdfs上的学生数据，统计每个班级的人数，写回到hdfs上
object Demo19YarnCluster {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("spark yarn cluster")
    val sc: SparkContext = new SparkContext(conf)

    //如果是打包到集群的话，这里的路径就是hdfs路径
    //如果是local的话，这个路径就是我们windows的路径
    val linesRDD: RDD[String] = sc.textFile("/bigdata30/students.csv")

    //coalesce函数后面会详细，可以修改分区
//    val linesRDD2: RDD[String] = linesRDD.coalesce(1)
//    linesRDD.repartition(1)

    println("=============================================================================================================")
    println(s"========================== linesRDD的分区数是:${linesRDD.getNumPartitions} ===================================")
    println("=============================================================================================================")

    val clazzKVRDD: RDD[(String, Int)] = linesRDD.map((line: String) => {
      line.split(",") match {
        case Array(_, _, _, _, clazz: String) =>
          (clazz, 1)
      }
    })

    val resultRDD: RDD[(String, Int)] = clazzKVRDD.reduceByKey(_ + _)

    val resultRDD2: RDD[String] = resultRDD.map((t2: (String, Int)) => s"${t2._1},${t2._2}")
    println("=============================================================================================================")
    println(s"========================== resultRDD2的分区数是:${resultRDD2.getNumPartitions} ===================================")
    println("=============================================================================================================")

    //行动算子，触发作业执行
    resultRDD2.saveAsTextFile("/bigdata30/sparkout1")

  }
}

/**
 * spark-submit --class com.shujia.core.Demo19YarnCluster --master yarn --deploy-mode cluster spark-1.0.jar
 */
